package com.alison.datastream.chapter4_join;

import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class F4_CoGroupExample2 {

    // CoGroup算子：将两个数据流按照key进行group分组，并将数据流按key进行分区的处理，最终合成一个数据流（与join有区别，不管key有没有关联上,最终都会合并成一个数据流）
    /*
    left data
    nc -lk 9999
    101,Tom
    102,小明
    103,小黑
    104,张强
    105,Ken
    106,GG小日子
    107,小花
    108,赵宣艺
    109,明亮

    right data
    nc -lk 9998
    101,男,本科,程序员
    102,男,本科,程序员
    103,女,本科,会计
    104,男,大专,安全工程师
    105,男,硕士,律师
    106,未知,小本,挖粪使者
    108,女,本科,人事
    110,男,本科,算法工程师
     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 使用EventTime时间语义
        env.setParallelism(1);
        // 如果配置了event_time，就配置watermark
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

//        List<Tuple3<String, String, Integer>> tuple3List1 = Arrays.asList(
//                new Tuple3<>("李四", "girl", 24),
//                new Tuple3<>("刘六", "girl", 32)
//        );
//        List<Tuple3<String, String, Integer>> tuple3List2 = Arrays.asList(
//                new Tuple3<>("伍七", "girl", 18),
//                new Tuple3<>("吴八", "man", 30)
//        );
        // 无法触发
//        DataStreamSource<Tuple3<String, String, Integer>> input1 = env.fromCollection(tuple3List1);
//        DataStreamSource<Tuple3<String, String, Integer>> input2 = env.fromCollection(tuple3List2);

        DataStream<Tuple2<Integer, String>> input1 = env.addSource(
                        new MySource(Lists.newArrayList(
                                "101,Tom",
                                "102,小明",
                                "103,小黑",
                                "104,张强",
                                "105,Ken",
                                "106,GG小日子",
                                "107,小花",
                                "108,赵宣艺",
                                "109,明亮")))
//        DataStream<Tuple2<Integer, String>> input1 = env.socketTextStream("192.168.56.101", 9999)
                .map(x -> {
                    String[] split = x.split(",");
                    return Tuple2.of(Integer.parseInt(split[0]), split[1]);
                })
                .returns(Types.TUPLE(Types.INT, Types.STRING));
        DataStream<Tuple4<Integer, String, String, String>> input2 = env.addSource(
                        new MySource(Lists.newArrayList(
                                "101,男,本科,程序员",
                                "102,男,本科,程序员",
                                "103,女,本科,会计",
                                "104,男,大专,安全工程师",
                                "105,男,硕士,律师",
                                "106,未知,小本,挖粪使者",
                                "108,女,本科,人事",
                                "110,男,本科,算法工程师")))
//        DataStream<Tuple4<Integer, String, String, String>> input2 = env.socketTextStream("192.168.56.101", 9998)
                .map(x -> {
                    String[] split = x.split(",");
                    return Tuple4.of(Integer.parseInt(split[0]), split[1], split[2], split[3]);
                })
                .returns(Types.TUPLE(Types.INT, Types.STRING, Types.STRING, Types.STRING));
        input1.print("input1");
        input2.print("input2");
        DataStream<String> coGroupResult = input1
                .coGroup(input2)
                .where(i1 -> i1.f0)// input1.key
                .equalTo(i2 -> i2.f0)// input2.key ,  where input1.key=input2.key
                .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))  // 开窗口,以处理时间划分(每20秒一个窗口)
                .apply(new CoGroupFunction<Tuple2<Integer, String>, Tuple4<Integer, String, String, String>, String>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<Integer, String>> first, Iterable<Tuple4<Integer, String, String, String>> second, Collector<String> out) throws Exception {
                        /*
                         * first 代表左流的迭代器
                         * second 代表右流的迭代器
                         * out 则是返回的数据形式
                         * 具体方法中两个迭代器存数据的原理后续会通过图结合进行解析
                         **/
                        // 这里的逻辑模拟sql中left join
                        // 遍历左流数据(first)
                        for (Tuple2<Integer, String> left : first) {
                            // 定义右流是否为NULL判断标识
                            boolean flag = false;
                            // 遍历右流数据(second)
                            for (Tuple4<Integer, String, String, String> right : second) {
                                // 返回left(id, name) + right(gender, education)
                                Tuple4<Integer, String, String, String> tuple4 = Tuple4.of(left.f0, left.f1, right.f2, right.f3);
                                out.collect(tuple4.toString());
                                // 修改判断标识
                                flag = true;
                            }
                            if (!flag) {
                                Tuple4<Integer, String, String, String> tuple4 = Tuple4.of(left.f0, left.f1, "NULL", "NULL");
                                out.collect(tuple4.toString());
                            }
                        }
                    }
                });
        coGroupResult.print("result");
        env.execute("cogroup function");
    }

    /**
     * IN1: Tuple2<String, Integer>
     * IN2: Tuple2<String, Integer>
     * OUT: String
     */
    public static class MyCoGroupFunction implements CoGroupFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String> {
        @Override
        public void coGroup(Iterable<Tuple2<String, Integer>> input1, Iterable<Tuple2<String, Integer>> input2, Collector<String> out) {
            input1.forEach(element -> System.out.println("input1 :" + element.f1));
            input2.forEach(element -> System.out.println("input2 :" + element.f1));
        }
    }

    public static class MySource implements SourceFunction<String> {
        private volatile boolean isRunning = true;

        int count = 0;

        List<String> content;

        public MySource(List<String> content) {
            this.content = content;
        }

        @Override
        public void run(SourceContext ctx) throws Exception {
            while (isRunning) {
                if (count == content.size()) {
//                    isRunning = false;
//                    continue;
                    count = 0;
                }
                ctx.collect(content.get(count));
                count++;
                TimeUnit.SECONDS.sleep(2);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
